package org.marsdonne.monitring_agent.activemq

import org.apache.activemq.broker.jmx.BrokerViewMBean
import org.apache.activemq.broker.jmx.QueueViewMBean
import org.marsdonne.monitring_agent.Reporter

import javax.management.MBeanServerConnection
import javax.management.MBeanServerInvocationHandler
import javax.management.ObjectName
import javax.management.remote.JMXConnector
import javax.management.remote.JMXConnectorFactory
import javax.management.remote.JMXServiceURL
import java.text.SimpleDateFormat

class JmxMonitor {
    final static dateFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    List<String> whitelist
    List<String> blacklist

    String connectorHost = "localhost"
    Integer connectorPort = 1099
    String connectorPath = "/jmxrmi"
    String jmxDomainName = "org.apache.activemq"

    private checkInterval = 180
    private boolean keepRunning = true

    private JMXConnector connector

    private Reporter reporter

    boolean enabledEmergencyRestart = false
    private Restarter restarter
    private int emergencyCounter = 0

    JmxMonitor() {
        final url = new JMXServiceURL("service:jmx:rmi:///jndi/rmi://$connectorHost:$connectorPort$connectorPath")
        connector = JMXConnectorFactory.connect(url)

        Thread.start {
            final Map<String, Checkable> thresholdMapping = [:]

            while (keepRunning) {
                try {
                    try {
                        connector.connect()
                    }
                    catch (IOException ex) {
                        connector = JMXConnectorFactory.connect(url)
                    }

                    MBeanServerConnection connection = connector.getMBeanServerConnection()

                    def name = new ObjectName(jmxDomainName + ":brokerName=localhost,type=Broker")

                    def mBean = MBeanServerInvocationHandler.newProxyInstance(connection, name, BrokerViewMBean.class, true)

                    if (mBean.getQueues()) {
                        def matchedQueues = 0

                        for (ObjectName queueName : mBean.getQueues()) {
                            def queueMBean = MBeanServerInvocationHandler.newProxyInstance(connection, queueName, QueueViewMBean.class, true)

                            if (
                            (blacklist.isEmpty() || !blacklist.contains(queueMBean.name))
                                    || (whitelist.isEmpty() || whitelist.contains(queueMBean.name))
                            ) {
                                def threshold = thresholdMapping.get(queueMBean.name, new QueueCongestionThreshold(queueName: queueMBean.name))

                                if (threshold.trigger(queueMBean)) {
                                    reporter.send("【警告】【${dateFormatter.format(new Date())}】消息队列[$queueMBean.name]已拥堵", threshold.warning, description)

                                    if (enabledEmergencyRestart) {
                                        emergencyCounter++

                                        if (emergencyCounter > 3) {
                                            reporter.send("【警告】【${dateFormatter.format(new Date())}】因消息队列[$queueMBean.name]拥堵，启动应急重启", restarter.process(), description)
                                            emergencyCounter = 0
                                        }
                                    }
                                }

                                matchedQueues++
                            }
                        }

                        if (!matchedQueues) {
                            reporter.send("黑/白名单不匹配", "监控失效，没有队列被规则匹配。", description)
                        }
                    } else {
                        reporter.send("没有消息队列", "没有可被监控的消息队列", description)
                    }

                    sleep checkInterval * 1000
                }
                catch (Exception ex) {
                    reporter.send("消息队列监控运行异常", ex.message, description)
                    sleep 10 * 1000
                }
            }
        }
    }

    String getDescription() {
        "\n监控说明\n监控白名单:$whitelist\n监控黑名单:$blacklist"
    }

    void stop() {
        keepRunning = false
        connector.close()
    }

    static interface Checkable {
        boolean trigger(QueueViewMBean queueMBean)

        String getWarning()
    }

    static class QueueCongestionThreshold implements Checkable {

        String queueName

        Long queueSize
        Long dequeueCount
        Long enqueueCount
        Long consumerCount

        @Override
        boolean trigger(QueueViewMBean queueMBean) {
            def result = false

            queueMBean.enqueueCount

            //非首次初始化
            if (queueSize) {
                //本次与上次的累积消费数的差值小于上次剩余量时，代表该队列已拥堵
                if ((queueMBean.dequeueCount - dequeueCount < queueSize) && (queueMBean.consumerCount > 0)) {
                    result = true
                }
                //有未消费消息、累积入列消息增长、有累积出列消息、但当前没有消费者时
                else if (queueMBean.consumerCount < 1 && queueMBean.enqueueCount > enqueueCount && queueMBean.dequeueCount) {
                    result = true
                }
            }

            queueSize = queueMBean.queueSize
            dequeueCount = queueMBean.dequeueCount
            enqueueCount = queueMBean.enqueueCount
            consumerCount = queueMBean.consumerCount

            return result
        }

        @Override
        String getWarning() {
            "消息队列:$queueName\n未消费（Pending）:$queueSize, 消费者数（Consumers）:$consumerCount, 累积入列（Enqueued）:$enqueueCount, 累积出列（Dequeued）:$dequeueCount"
        }
    }
}